package com.shujia.sql

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{Column, DataFrame, SparkSession}

object Demo5Burk {
  def main(args: Array[String]): Unit = {

    val spark: SparkSession = SparkSession.builder()
      .master("local")
      .appName("dianxin")
      .config("spark.sql.shuffle.partitions", 2)
      .getOrCreate()


    //导入隐式转换
    import spark.implicits._

    //导入sql 所有的函数
    import org.apache.spark.sql.functions._


    /**
      * explode: 相当于flatmap
      *
      */

    spark.sql(
      """
        |select explode(array(1,2,3,4,5,6))
        |
      """.stripMargin)
    // .show()


    spark.sql(
      """
        |select explode(map('01','张三','02','李四','03','王五'))
        |
      """.stripMargin)
    //.show()

    /**
      * 1、统计每个公司每年按月累计收入
      *
      * 输出结果
      * 公司代码,年度,月份,当月收入,累计收入
      *
      */

    val burk: DataFrame = spark.read
      .format("csv")
      .option("sep", ",")
      .schema("burk STRING,year STRING,tsl01 DOUBLE,tsl02 DOUBLE,tsl03 DOUBLE,tsl04 DOUBLE,tsl05 DOUBLE,tsl06 DOUBLE,tsl07 DOUBLE,tsl08 DOUBLE,tsl09 DOUBLE,tsl10 DOUBLE,tsl11 DOUBLE,tsl12 DOUBLE")
      .load("spark/data/burks.txt")

    burk.createOrReplaceTempView("burk")


    spark.sql(
      """
        |
        |select
        |burk,year,month,pic
        |from
        |burk
        |LATERAL VIEW  explode(map( 1, tsl01,2 ,tsl02, 3 , tsl03,4,tsl04,5, tsl05,6 ,tsl06, 7 , tsl07,8,tsl08,9, tsl09,10 ,tsl10, 11 , tsl11,12,tsl12)) t as month,pic
        |
      """.stripMargin)


    spark.sql(
      """
        |
        |select burk,year,month ,pic ,sum(pic) over (partition by burk,year order by month) as sumPic
        |from
        |(select
        |burk,year,explode(map( 1, tsl01,2 ,tsl02, 3 , tsl03,4,tsl04,5, tsl05,6 ,tsl06, 7 , tsl07,8,tsl08,9, tsl09,10 ,tsl10, 11 , tsl11,12,tsl12)) as (month,pic)
        |from
        |burk) as a
        |
      """.stripMargin)
    //.show(1000)


    /**
      * DSL
      *
      */

    val monthMap: Column = map(
      expr("1"), $"tsl01",
      expr("2"), $"tsl02",
      expr("3"), $"tsl03",
      expr("4"), $"tsl04",
      expr("5"), $"tsl05",
      expr("6"), $"tsl06",
      expr("7"), $"tsl07",
      expr("8"), $"tsl08",
      expr("9"), $"tsl09",
      expr("10"), $"tsl10",
      expr("11"), $"tsl11",
      expr("12"), $"tsl12"
    )


    burk
      .select($"burk", $"year", explode(monthMap) as Array("month", "pic"))
      .select($"burk", $"year", $"month", $"pic", sum($"pic") over Window.partitionBy($"burk", $"year").orderBy($"month".asc) as "sumPic")
      .show(1000)


    /**
      *
      * 2、统计每个公司当月比上年同期增长率
      * 公司代码,年度,月度,增长率（当月收入/上年当月收入 - 1）
      *
      */

    /**
      * coalesce ： 如第一个为null 再取第二个
      *
      */

    burk
      .select($"burk", $"year", explode(monthMap) as Array("month", "pic"))
      .select($"burk", $"year", $"month", $"pic", lag($"pic", 1, 0.0) over Window.partitionBy($"burk", $"month").orderBy($"year") as "lastPic")
      .select($"burk", $"year", $"month", $"pic", round(coalesce($"pic" / $"lastPic" - 1, expr("1.0")), 5) as "bi")
      .show()

  }

}
